package com.atguigu.kafka.partitioner;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;

/**
 * 自定义分区器
 *
 * 需求: 消息中包含"atguigu"，发送到0号分区
 *       其他的消息发送到1号分区
 */
public class MyPartitioner  implements Partitioner {
    /**
     * 计算分区号
     * @param topic  当前消息发送到的主题
     * @param key    消息的key
     * @param keyBytes 序列化以后的key
     * @param value  消息
     * @param valueBytes 序列化后的消息
     * @param cluster
     * @return
     */
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        if(value.toString().contains("atguigu")){
            return 0 ;
        }
        return 1;
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
